// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <functional>
#include <map>
#include <set>
#include <string>
#include <string_view>
#include <vector>

#include "absl/container/flat_hash_map.h"

#include <glog/logging.h>

#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/executor/lifecycle_timestamp.h"

namespace coin2::exchange::base::feed {

using coin2::exchange::base::executor::LifecycleTimestamp;

class FeedDistributor {
 public:
  FeedDistributor() {}
  FeedDistributor(LifecycleTimestamp* ts) : ts_(ts) {}
  const std::set<std::string, std::less<>>& native_symbols() const {
    return native_symbols_;
  }

  bool is_symbol_registered(std::string_view native_symbol) const {
    return native_symbols_.count(native_symbol) > 0;
  }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    const auto& native_symbol = product.feed_native_symbol();
    CHECK(!native_symbol.empty()) << native_symbol;
    CHECK(subscriber) << subscriber;
    CHECK_EQ(native_symbols_.count(native_symbol), 0) << native_symbol;
    CHECK_EQ(subscriber_map_.count(native_symbol), 0) << native_symbol;
    native_symbols_.insert(native_symbol);
    subscriber_map_[native_symbol] = subscriber;
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    for (const auto& product : products) {
      const auto& native_symbol = product->feed_native_symbol();
      if (native_symbols_.count(native_symbol) == 0) {
        LOG(INFO) << "Symbol not yet subscribed " << native_symbol;
      };
      native_symbols_.erase(native_symbol);
      subscriber_map_.erase(native_symbol);
    }
  }

  virtual void onTradeFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onTradeFeed(upd);
    }
  }

  virtual void onBookFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onBookFeed(upd);
    }
  }

  virtual void onLiquidationFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onLiquidationFeed(upd);
    }
  }

  virtual void onOpenInterestFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onOpenInterestFeed(upd);
    }
  }

  virtual void onMarkPriceFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onMarkPriceFeed(upd);
    }
  }

  virtual void onIndexFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onIndexFeed(upd);
    }
  }

  virtual void onFundingRateFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onFundingRateFeed(upd);
    }
  }

  virtual void onIvGreeksFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onIvGreeksFeed(upd);
    }
  }

  virtual void onNavFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onNavFeed(upd);
    }
  }

  virtual void onKlineFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onKlineFeed(upd);
    }
  }

  virtual void onWalletFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onWalletFeed(upd);
    }
  }

  virtual void onMarketAnalysisFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onMarketAnalysisFeed(upd);
    }
  }

  virtual void onTopLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onTopLongShortAccountRatioFeed(upd);
    }
  }

  virtual void onTopLongShortPositionRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onTopLongShortPositionRatioFeed(upd);
    }
  }

  virtual void onGlobalLongShortAccountRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onGlobalLongShortAccountRatioFeed(upd);
    }
  }

  virtual void onTakerBuySellRatioFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onTakerBuySellRatioFeed(upd);
    }
  }

  virtual void onOpenInterestHistFeed(std::string_view native_symbol, const FeedUpdate& upd) {
    if (ts_) {
      ts_->set_main_tx_timestamp(GetCurrentTimestamp());
    }
    auto iter = subscriber_map_.find(native_symbol);
    if (iter != subscriber_map_.end()) {
      iter->second->onOpenInterestHistFeed(upd);
    }
  }

 private:
  std::set<std::string, std::less<>> native_symbols_;
  // std::map<std::string, IFeedSubscriber*, std::less<>> subscriber_map_;
  absl::flat_hash_map<std::string, IFeedSubscriber*> subscriber_map_;
  LifecycleTimestamp* ts_ = nullptr;
};

}  // namespace coin2::exchange::base::feed
